package com.github.ghsea.scheduler.util;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;


//@Service("kafkaService")
@Slf4j
public class KafkaService {

    @Resource(name = "kafkaTemplate")
    private KafkaTemplate<String, String> kafkaTemplate;

    public Future sendAsync(String topic, Integer partition, String key, String data, Callback callback) {
        Producer<String, String> producer = kafkaTemplate.getProducerFactory().createProducer();
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, partition, key, data);
        Future<RecordMetadata> send = producer.send(producerRecord, (recordMetadata, e) -> {
            try {
                producer.close();
            } finally {
                log.debug("recordMetadata the topic :{}, key :{},offset:{}", recordMetadata.topic(), key, recordMetadata.offset());
                if (e != null) {
                    log.error("kafka topic:{} msg sent failed:{}", recordMetadata.topic(), data, e);
                } else {
                    log.debug("kafka topic:{} msg sent success:{}", recordMetadata.topic(), data);
                }
                if (callback != null) {
                    callback.action(new SendResult<>(producerRecord, recordMetadata), e);
                }
            }
        });
        return send;
    }

    public Future sendAsync(String topic, Integer partition, String key, String data) {
        return sendAsync(topic, partition, key, data, null);
    }

    public Future sendAsync(String topic, String key, String data) {
        return sendAsync(topic, null, key, data, null);
    }

    public void sendAsync(String topic, String key, String data, Callback callback) {
        sendAsync(topic, null, key, data, callback);
    }

    public RecordMetadata send(String topic, Integer partition, String key, String data) {
        Producer<String, String> producer = kafkaTemplate.getProducerFactory().createProducer();
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, partition, key, data);
        RecordMetadata future = null;
        try {
            future = producer.send(producerRecord).get();
        } catch (InterruptedException e) {
            log.error("【KafkaService.send】Exception", e);
        } catch (ExecutionException e) {
            log.error("【KafkaService.send】Exception", e);
            throw new RuntimeException(e);
        } finally {
            producer.close();
        }
        return future;
    }

    public void send(String topic, String key, String data) {
        send(topic, null, key, data);
    }

    public interface Callback {
        void action(SendResult<String, String> sendResult, Throwable throwable);
    }

}

